package com.ypj.rpc.transport.netty.server;

import com.ypj.rpc.entity.RpcRequest;
import com.ypj.rpc.entity.RpcResponse;
import com.ypj.rpc.factory.SingletonFactory;
import com.ypj.rpc.handler.RequestHandler;
import com.ypj.rpc.factory.ThreadPoolFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;

@Slf4j
public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final String THREAD_NAME_PREFIX="netty-server-handler";
    private final ExecutorService threadPool;
    private final RequestHandler requestHandler;

    public NettyServerHandler() {
        //生成单例的handler
        this.requestHandler = SingletonFactory.getInstance(RequestHandler.class);
        this.threadPool= ThreadPoolFactory.createDefaultThreadPool(THREAD_NAME_PREFIX);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
        //引入异步业务线程池的发你说个事避免长时间业务阻塞netty本身的worker线程
        threadPool.execute(()->{
            try {
                if (msg.getHeartBeat()){
                    log.info("接收到客户端心跳包...");
                    return ;
                }
                log.info("服务器接收到请求：{}",msg);
                Object result=requestHandler.handle(msg);
                ChannelFuture future = ctx.writeAndFlush(RpcResponse.success(result, msg.getRequestId()));
                future.addListener(ChannelFutureListener.CLOSE);
            } finally {
                ReferenceCountUtil.release(msg);
            }
        });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("调用过程发生故障：");
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleState state = ((IdleStateEvent) evt).state();
            if (state==IdleState.READER_IDLE){
                log.info("长时间未收到心跳包，断开连接...");
                ctx.close();
            }
        }else{
            super.userEventTriggered(ctx,evt);
        }
    }
}
